{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Env"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Set base path\n",
    "BASE_PATH = \"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import sys\n",
    "sys.path.append(os.path.abspath(\"..\"))\n",
    "\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql import types as T\n",
    "from pyspark.sql.functions import col, countDistinct, count, desc, broadcast, lower, sum, row_number, floor\n",
    "from pyspark.sql.window import Window\n",
    "\n",
    "from helpers.variables import COLS_TAXONOMIC\n",
    "from helpers.data_analysis import init_spark, create_freq, view_freq, check_sparsity\n",
    "from helpers.gbif import fetch_gbif_chunk, fetch_gbif_iter, retry_failed_chunks, insert_records_to_mongo, fetch_publisher_key\n",
    "#from helpers.text_search import flatten_dict, full_text_search_rdd, flatten_list_to_string, extract_fields\n",
    "\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# bash:\n",
    "# mongod --dbpath=\"${BASE_PATH}/gbif/gbif_mongo\" --fork --logpath=\"${BASE_PATH}/mongo_logs/gbif_mongo.log\"\n",
    "# Check process status\n",
    "# ps aux | grep mongod\n",
    "\n",
    "from pymongo import MongoClient\n",
    "\n",
    "# Connect to the MongoDB server\n",
    "client = MongoClient(\"mongodb://localhost:27017/\")\n",
    "\n",
    "db = client[\"gbif\"] \n",
    "collection_registry = db[\"registry\"]  \n",
    "collection_grscicoll_collection = db[\"grscicoll_collection\"]  \n",
    "collection_grscicoll_institution = db[\"grscicoll_institution\"] \n",
    "collection_organization = db[\"organization\"] "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Download GRSciColl Collection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "base_url = \"https://api.gbif.org/v1/grscicoll/collection\"\n",
    "all_records_collection, all_logs_collection = fetch_gbif_iter(base_url, params=None, limit=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Inserting records: 100%|██████████| 8928/8928 [00:03<00:00, 2948.26doc/s]\n"
     ]
    }
   ],
   "source": [
    "insert_records_to_mongo(\n",
    "    collection_grscicoll_collection,\n",
    "    all_records_collection,\n",
    "    unique_key = \"key\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "base_url = \"https://api.gbif.org/v1/grscicoll/institution\"\n",
    "all_records_institution, all_logs_institution = fetch_gbif_iter(base_url, params=None, limit=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Inserting records: 100%|██████████| 8832/8832 [00:03<00:00, 2890.41doc/s]\n"
     ]
    }
   ],
   "source": [
    "insert_records_to_mongo(\n",
    "    collection_grscicoll_institution,\n",
    "    all_records_institution,\n",
    "    unique_key = \"key\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Download organization"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "base_url = \"https://api.gbif.org/v1/organization\"\n",
    "all_records_organization, all_logs_organization = fetch_gbif_iter(base_url, params=None, limit=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Inserting records: 100%|██████████| 2980/2980 [00:01<00:00, 2762.60doc/s]\n"
     ]
    }
   ],
   "source": [
    "insert_records_to_mongo(\n",
    "    collection_organization,\n",
    "    all_records_organization,\n",
    "    unique_key = \"key\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Download occurrence registery"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "base_url = \"https://api.gbif.org/v1/dataset\"\n",
    "all_records, all_logs = fetch_gbif_iter(base_url, params = None, limit=500)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "MongoDB uses the `_id` field as a unique identifier for each document. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "49968"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "collection.count_documents({})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Approach Summary"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Marie Grosjean is the Data Adminstrator from GBIF. She developed an [NLP approach](https://data-blog.gbif.org/post/gbif-citizen-science/) to identify and automatically label datasets as citizen science (CS) using the metadata available via the [GBIF Dataset API](https://techdocs.gbif.org/en/openapi/v1/registry#/). Marie tagged these datasets in the API using a `machineTag` and there are [638 identified dataset matches](http://api.gbif.org/v1/dataset?machineTagNamespace=citizenScience.gbif.org).\n",
    "\n",
    "We've noticed that the NLP model has failed to capture many citizen science datasets and also mislabeled some non-CS dataset, such as [AntWeb](https://www.gbif.org/dataset/13b70480-bd69-11dd-b15f-b8a03c50a862) ([JSON](https://api.gbif.org/v1/dataset/13b70480-bd69-11dd-b15f-b8a03c50a862)).\n",
    "\n",
    "We developed an rule-based full-text search approach to increase the precision and coverage:\n",
    "\n",
    "- **Scope of search:** \n",
    "    - Specific multilingual (English, Spanish, Portuguese, French) keywords (singular & plural) for \"citizen\" and \"citizen science\"\n",
    "    - Known and verified CS publishers such as iNaturalist, Observation.org, etc.\n",
    "    - Known and verified CS datasets\n",
    "- **Preprocessing**\n",
    "    - Remove all machineTags to prevent introducing results from Marie's NLP model's prior prediction\n",
    "    - No additional text cleaning steps applied\n",
    "- **Pros & Cons**\n",
    "    - Pros: Fast, scalable, reproducible, deterministic\n",
    "    - Cons: May introduce recall by missing datasets that do not contain the predefined search terms. May also introduce non-CS datasets that contains seach terms in the metadata."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "base_url = \"https://api.gbif.org/v1/dataset?machineTagNamespace=citizenScience.gbif.org\"\n",
    "all_records, all_logs = fetch_gbif_iter(base_url, params = None, limit=500)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "key_values = [record['key'] for record in all_records if 'key' in record]\n",
    "matching_df_nlp = spark.createDataFrame([(value,) for value in key_values], [\"datasetKey\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark full-text search"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Check if there is an active Spark session\n",
    "spark= SparkSession.getActiveSession()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark = (SparkSession.builder\n",
    "             .appName(\"GBIF EDA\")\n",
    "             .config(\"spark.executor.instances\", \"64\")\n",
    "             .config(\"spark.executor.memory\", \"75G\")\n",
    "             .config(\"spark.executor.cores\", \"12\")\n",
    "             .config(\"spark.sql.parquet.enableVectorizedReader\", \"false\") \n",
    "             .getOrCreate())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sparkContext.addPyFile(os.path.abspath(\"../helpers/text_search.py\"))\n",
    "\n",
    "from text_search import flatten_dict, full_text_search_rdd, flatten_list_to_string, extract_fields"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "publisher_citizen_sci = [\n",
    "    \"iNaturalist.org\",\n",
    "    \"Observation.org\",\n",
    "    \"naturgucker.de\",\n",
    "    \"Questagame\",\n",
    "    \"Pl@ntNet\",\n",
    "    \"NatureMapr\",\n",
    "    \"Citizen Science - ALA Website\",\n",
    "    \"BioCollect\",\n",
    "    \"Tweed Koala Sightings\",\n",
    "    \"Koala Action Group\",\n",
    "    \"TilapiaMap\",\n",
    "    \"Blauwtipje.nl\",\n",
    "    \"Great Koala Count 2\",\n",
    "    #\"myFOSSIL eMuseum\",\n",
    "    \"Superb Parrot Monitoring project\",\n",
    "    \"SLU Artdatabanken\",\n",
    "    \"Sibecocenter LLC\"\n",
    "\n",
    "]\n",
    "\n",
    "publisher_citizen_sci_df = pd.DataFrame(fetch_publisher_key(publisher_citizen_sci))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "datasetKey_citizen_sci = [\n",
    "    \"84a649ce-ff81-420d-9c41-aa1de59e3766\", # Citizen Science - ALA Website\n",
    "    \"cca13f2c-0d2c-4c2f-93b9-4446c0cc1629\"  # BugGuide published by United States Geological Survey\n",
    "]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "search_terms = [\n",
    "    # 'citizen' in different languages, in singular and plural\n",
    "    \"citizen\", \"citizens\",\n",
    "    # ciencia ciencias\n",
    "    \"ciudadana\", \"ciudadano\", # feminine masculine\n",
    "    \"ciudadanas\", \"ciudadanos\",\n",
    "    \"cidadã\", \"cidadãs\",\n",
    "    \"citoyenne\", \"citoyennes\",\n",
    "    \n",
    "    #\"inaturalist\", \"observation.org\",\n",
    "    \n",
    "    # 'citizen science' in different languages, in singular and plural\n",
    "    \"citizen science\", \"citizen-science\",\n",
    "    \"ciencia ciudadana\", # Spanish\n",
    "    \"ciência cidadã\",    # Portuguese\n",
    "    \"science citoyenne\"  # French\n",
    "]\n",
    "\n",
    "search_terms.extend(list(pd.DataFrame(publisher_citizen_sci_df)[\"key\"])) # keys of the known citizen science publishers\n",
    "search_terms.extend(datasetKey_citizen_sci)                              # keys of citizen science datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 8.54 s, sys: 1.82 s, total: 10.4 s\n",
      "Wall time: 11.5 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "rdd = spark.sparkContext.parallelize(\n",
    "    list(collection_registry.find({}))\n",
    "    # list(collection.find({}, {\"machineTags\": 0})) # Exclude the machineTag from the downstream full text search\n",
    ")\n",
    "\n",
    "rdd_filtered = rdd.map(lambda record: {k: v for k, v in record.items() if k != \"machineTags\"})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd = None"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Without `machineTags`**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:46:39 WARN TaskSetManager: Stage 0 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: citizen, Count: 353\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:46:43 WARN TaskSetManager: Stage 1 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n",
      "25/01/08 16:46:46 WARN TaskSetManager: Stage 2 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: citizens, Count: 63\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:46:49 WARN TaskSetManager: Stage 3 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: ciudadana, Count: 36\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:46:51 WARN TaskSetManager: Stage 4 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: ciudadano, Count: 34\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:46:53 WARN TaskSetManager: Stage 5 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: ciudadanas, Count: 0\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:46:56 WARN TaskSetManager: Stage 6 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: ciudadanos, Count: 13\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:46:58 WARN TaskSetManager: Stage 7 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: cidadã, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:01 WARN TaskSetManager: Stage 8 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: cidadãs, Count: 0\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:03 WARN TaskSetManager: Stage 9 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: citoyenne, Count: 1925\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:06 WARN TaskSetManager: Stage 10 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: citoyennes, Count: 16\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:08 WARN TaskSetManager: Stage 11 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: citizen science, Count: 271\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:10 WARN TaskSetManager: Stage 12 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: citizen-science, Count: 8\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:13 WARN TaskSetManager: Stage 13 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: ciencia ciudadana, Count: 23\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:15 WARN TaskSetManager: Stage 14 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: ciência cidadã, Count: 0\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:18 WARN TaskSetManager: Stage 15 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: science citoyenne, Count: 1897\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:20 WARN TaskSetManager: Stage 16 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: 0cab7ac5-e262-4cb5-b6ec-2f524c1f7975, Count: 0\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:23 WARN TaskSetManager: Stage 17 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: 28eb1a3f-1c15-4a95-931a-4af90ecb574d, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:25 WARN TaskSetManager: Stage 18 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: c8d737e0-2ff8-42e8-b8fc-6b805d26fc5f, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:27 WARN TaskSetManager: Stage 19 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: bb646dff-a905-4403-a49b-6d378c2cf0d9, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:30 WARN TaskSetManager: Stage 20 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: 17726825-3aa2-4fbc-b0e3-303d3fc5702f, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:32 WARN TaskSetManager: Stage 21 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: da86174a-a605-43a4-a5e8-53d484152cd3, Count: 2\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:35 WARN TaskSetManager: Stage 22 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: d9bea9d3-13a5-4768-bbf4-560b9aa95a73, Count: 2\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:37 WARN TaskSetManager: Stage 23 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: adc174cd-c752-4eee-9630-7c1209eb1c4a, Count: 0\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:39 WARN TaskSetManager: Stage 24 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: 1d9d1d61-7871-44b4-9ceb-0bc2fe809d2e, Count: 26\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:42 WARN TaskSetManager: Stage 25 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: 12379077-6d25-42da-a88d-5f4916e5bba9, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:44 WARN TaskSetManager: Stage 26 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: facf3dd1-4b89-4780-ba49-b5fdef24c7d3, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:47 WARN TaskSetManager: Stage 27 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: 5efc1e1a-ba2f-471e-a8ed-e5849225e949, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:49 WARN TaskSetManager: Stage 28 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: f987a53f-bbe7-4a5e-b956-47187dbe3069, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:52 WARN TaskSetManager: Stage 29 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: f48f13a4-84e9-4b13-bf97-cc387d3c3229, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:54 WARN TaskSetManager: Stage 30 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: 990eac22-e9a5-4499-baca-3e79e5cabd41, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:56 WARN TaskSetManager: Stage 31 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: b8323864-602a-4a7d-9127-bb903054e97d, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:47:59 WARN TaskSetManager: Stage 32 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: 477dbd45-674f-4389-8952-7c9bbde10f68, Count: 3\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/08 16:48:01 WARN TaskSetManager: Stage 33 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: 84a649ce-ff81-420d-9c41-aa1de59e3766, Count: 1\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 33:=====================================================>(382 + 2) / 384]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Term: cca13f2c-0d2c-4c2f-93b9-4446c0cc1629, Count: 1\n",
      "CPU times: user 230 ms, sys: 149 ms, total: 379 ms\n",
      "Wall time: 1min 24s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "term_counts = []\n",
    "for term in search_terms:\n",
    "    matching_rdd = full_text_search_rdd(rdd_filtered, term)\n",
    "    count = matching_rdd.count() \n",
    "    print(f\"Term: {term}, Count: {count}\")\n",
    "    term_counts.append({\"Term\": term, \"Count\": count})  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/02/05 15:31:11 WARN TaskSetManager: Stage 0 contains a task of very large size (1579 KiB). The maximum recommended task size is 1000 KiB.\n",
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "2378"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Perform search all terms\n",
    "matching_rdd = full_text_search_rdd(rdd_filtered, search_terms)\n",
    "\n",
    "match_count = matching_rdd.count()\n",
    "#results = matching_rdd.collect()\n",
    "match_count"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "25/01/14 12:02:57 WARN TaskSetManager: Stage 2 contains a task of very large size (1061 KiB). The maximum recommended task size is 1000 KiB.\n",
      "[Stage 2:======================================================>(767 + 1) / 768]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 15.3 ms, sys: 25.4 ms, total: 40.6 ms\n",
      "Wall time: 6.51 s\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "2378"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "fields_to_extract = [\"_id\", \"publishingOrganizationKey\"]\n",
    "extract_fields_func = extract_fields(fields_to_extract)\n",
    "\n",
    "matching_df = spark.createDataFrame(matching_rdd.map(extract_fields_func)).distinct()\n",
    "matching_df.cache()\n",
    "\n",
    "matching_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Free memory\n",
    "rdd_filtered = None \n",
    "matching_rdd = None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructField, StringType, StructType\n",
    "rdd_grscicoll_institution_all = spark.sparkContext.parallelize(\n",
    "    list(collection_grscicoll_institution.find({}))\n",
    ")\n",
    "\n",
    "fields_to_extract = [\n",
    "    \"_id\", \n",
    "    \"name\",\n",
    "    \"code\",\n",
    "    \"types\",\n",
    "    \"institutionalGovernances\",      # Instutional governance of a GrSciColl institution\n",
    "    \"disciplines\",                   # Discipline of a GrSciColl institution. Accepts multiple values, for example\n",
    "    \"masterSource\"                   # DATASET, ORGANIZATION\n",
    "]\n",
    "\n",
    "extract_fields_func = extract_fields(fields_to_extract, fields_to_flatten = [\"types\", \"institutionalGovernances\", \"disciplines\"])\n",
    "\n",
    "schema = StructType([\n",
    "    StructField(\"_id\", StringType(), True),\n",
    "    StructField(\"name\", StringType(), True),\n",
    "    StructField(\"code\", StringType(), True),\n",
    "    StructField(\"types\", StringType(), True),\n",
    "    StructField(\"institutionalGovernances\", StringType(), True),\n",
    "    StructField(\"disciplines\", StringType(), True),  # Flattened to String\n",
    "    StructField(\"masterSource\", StringType(), True)\n",
    "])\n",
    "\n",
    "grscicoll_institution_df = spark.createDataFrame(\n",
    "    rdd_grscicoll_institution_all.map(extract_fields_func),\n",
    "    schema\n",
    ")\n",
    "\n",
    "grscicoll_institution_type_list = rdd_grscicoll_institution_all.map(extract_fields([\"types\"]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _id: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- code: string (nullable = true)\n",
      " |-- types: string (nullable = true)\n",
      " |-- institutionalGovernances: string (nullable = true)\n",
      " |-- disciplines: string (nullable = true)\n",
      " |-- masterSource: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "grscicoll_institution_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------+--------------------+\n",
      "|column_name             |sparsity            |\n",
      "+------------------------+--------------------+\n",
      "|_id                     |0.0                 |\n",
      "|name                    |0.0                 |\n",
      "|code                    |0.004981884057971014|\n",
      "|types                   |0.5180027173913043  |\n",
      "|institutionalGovernances|0.9039855072463768  |\n",
      "|disciplines             |0.9394248188405797  |\n",
      "|masterSource            |0.0                 |\n",
      "+------------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "check_sparsity(grscicoll_institution_df).show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('UniversityCollege', 289),\n",
       " ('MuseumHerbariumPrivateNonProfit', 9),\n",
       " ('Herbarium', 3205),\n",
       " ('MedicalResearchInstitute', 7),\n",
       " ('OtherTypeResearchInstitutionBiorepository', 89),\n",
       " ('BiomedicalResearchInstitute', 10),\n",
       " ('Museum', 438),\n",
       " ('OtherInstitutionalType', 196),\n",
       " ('ZooAquarium', 10),\n",
       " ('BotanicalGarden', 63)]"
      ]
     },
     "execution_count": 110,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "create_freq_rdd(grscicoll_institution_type_list, key=\"types\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdd_grscicoll_collection_all = spark.sparkContext.parallelize(\n",
    "    list(collection_grscicoll_collection.find({}))\n",
    ")\n",
    "\n",
    "fields_to_extract = [\n",
    "    \"_id\", \n",
    "    \"name\",\n",
    "    \"code\",\n",
    "    \"contentTypes\",\n",
    "    \"preservationTypes\",\n",
    "    \"institutionKey\",      \n",
    "    \"institutionName\",                 \n",
    "    \"institutionCode\",\n",
    "    \"occurrenceCount\"\n",
    "]\n",
    "\n",
    "extract_fields_func = extract_fields(fields_to_extract, fields_to_flatten = [\"contentTypes\", \"preservationTypes\"])\n",
    "\n",
    "grscicoll_collection_df = spark.createDataFrame(\n",
    "    rdd_grscicoll_collection_all.map(extract_fields_func)\n",
    ")\n",
    "\n",
    "grscicoll_collection_contentTypes_list = rdd_grscicoll_institution_all.map(extract_fields([\"contentTypes\"]))\n",
    "grscicoll_collection_preservationTypes_list = rdd_grscicoll_institution_all.map(extract_fields([\"preservationTypes\"]))\n",
    "rdd_grscicoll_collection_all = None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _id: string (nullable = true)\n",
      " |-- code: string (nullable = true)\n",
      " |-- contentTypes: string (nullable = true)\n",
      " |-- institutionCode: string (nullable = true)\n",
      " |-- institutionKey: string (nullable = true)\n",
      " |-- institutionName: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- occurrenceCount: long (nullable = true)\n",
      " |-- preservationTypes: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "grscicoll_collection_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+--------------------+\n",
      "|column_name      |sparsity            |\n",
      "+-----------------+--------------------+\n",
      "|_id              |0.0                 |\n",
      "|code             |0.015568996415770609|\n",
      "|contentTypes     |0.6937724014336918  |\n",
      "|institutionCode  |0.02128136200716846 |\n",
      "|institutionKey   |0.00739247311827957 |\n",
      "|institutionName  |0.00739247311827957 |\n",
      "|name             |0.0                 |\n",
      "|occurrenceCount  |0.8538306451612904  |\n",
      "|preservationTypes|0.7720654121863799  |\n",
      "+-----------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "check_sparsity(grscicoll_collection_df).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Find match in occurrence"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark_df = spark.read.parquet(f\"{BASE_PATH}/gbif/attributes/cols_of_interest\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "filtered_df = (\n",
    "    spark_df\n",
    "    .join(\n",
    "        broadcast(matching_df), \n",
    "        matching_df._id == spark_df.datasetKey,\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Match count stats"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "107"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filtered_df.select(\"datasetKey\").distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "58"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filtered_df.select(\"publisher\").distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "110323916"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filtered_df.select(\"source_id\").distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+---------+\n",
      "|      basisOfRecord|    count|\n",
      "+-------------------+---------+\n",
      "|  HUMAN_OBSERVATION|181826921|\n",
      "| PRESERVED_SPECIMEN| 58611705|\n",
      "|    MATERIAL_SAMPLE|  3272814|\n",
      "|MACHINE_OBSERVATION|  3044865|\n",
      "|    FOSSIL_SPECIMEN|  2001234|\n",
      "|    LIVING_SPECIMEN|    99740|\n",
      "|         OCCURRENCE|    92770|\n",
      "|        OBSERVATION|    35199|\n",
      "|  MATERIAL_CITATION|     1492|\n",
      "+-------------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark_df.groupBy(\"basisOfRecord\").count().orderBy(desc(\"count\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+---------+\n",
      "|      basisOfRecord|    count|\n",
      "+-------------------+---------+\n",
      "|  HUMAN_OBSERVATION|179199017|\n",
      "|    MATERIAL_SAMPLE|   174405|\n",
      "| PRESERVED_SPECIMEN|   136773|\n",
      "|MACHINE_OBSERVATION|    48093|\n",
      "|        OBSERVATION|    17002|\n",
      "|    FOSSIL_SPECIMEN|    14087|\n",
      "|         OCCURRENCE|     1859|\n",
      "+-------------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "filtered_df.groupBy(\"basisOfRecord\").count().orderBy(desc(\"count\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Source ID multiple images subset"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check on subset source ID with multiple images"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "col_list = [\"source_id\", \"publisher\", \"basisOfRecord\", \"datasetKey\"]\n",
    "source_id_with_multiple_uuids = (\n",
    "    spark_df.groupBy(*col_list)\n",
    "    .agg(\n",
    "        F.countDistinct(\"uuid\").alias(\"distinct_uuid_count\"),\n",
    "    )\n",
    "    .filter(col(\"distinct_uuid_count\") > 1)\n",
    ")\n",
    "#source_id_with_multiple_uuids.cache()\n",
    "#source_id_with_multiple_uuids.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+\n",
      "|sum_value|\n",
      "+---------+\n",
      "|121343610|\n",
      "+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "source_id_with_multiple_uuids.agg(sum(\"distinct_uuid_count\").alias(\"sum_value\")).show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "filtered_df = (\n",
    "    source_id_with_multiple_uuids\n",
    "    .join(\n",
    "        broadcast(matching_df), \n",
    "        matching_df._id == source_id_with_multiple_uuids.datasetKey,\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "36404253"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filtered_df.select(\"source_id\").distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------+-------+\n",
      "|basisOfRecord      |count   |bucket |\n",
      "+-------------------+--------+-------+\n",
      "|HUMAN_OBSERVATION  |36365160|10m+   |\n",
      "|PRESERVED_SPECIMEN |26239   |10k-50k|\n",
      "|MATERIAL_SAMPLE    |10847   |10k-50k|\n",
      "|FOSSIL_SPECIMEN    |1655    |1k-5k  |\n",
      "|MACHINE_OBSERVATION|232     |101-500|\n",
      "|OCCURRENCE         |117     |101-500|\n",
      "|OBSERVATION        |3       |1-10   |\n",
      "+-------------------+--------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "view_freq(filtered_df, \"basisOfRecord\", truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "(\n",
    "    spark_df\n",
    "    .join(\n",
    "        source_id_with_multiple_uuids.filter(cond_camera_trap),\n",
    "        on = \"source_id\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    "    .count()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------+\n",
      "|      basisOfRecord|   count|\n",
      "+-------------------+--------+\n",
      "|  HUMAN_OBSERVATION|37227100|\n",
      "| PRESERVED_SPECIMEN| 3489859|\n",
      "|    FOSSIL_SPECIMEN|  276909|\n",
      "|MACHINE_OBSERVATION|  235632|\n",
      "|    MATERIAL_SAMPLE|  212826|\n",
      "|    LIVING_SPECIMEN|   19681|\n",
      "|        OBSERVATION|    3992|\n",
      "|         OCCURRENCE|    2998|\n",
      "|  MATERIAL_CITATION|     254|\n",
      "+-------------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "source_id_with_multiple_uuids.groupBy(\"basisOfRecord\").count().orderBy(desc(\"count\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------+\n",
      "|      basisOfRecord|   count|\n",
      "+-------------------+--------+\n",
      "|  HUMAN_OBSERVATION|36365160|\n",
      "| PRESERVED_SPECIMEN|   26239|\n",
      "|    MATERIAL_SAMPLE|   10847|\n",
      "|    FOSSIL_SPECIMEN|    1655|\n",
      "|MACHINE_OBSERVATION|     232|\n",
      "|         OCCURRENCE|     117|\n",
      "|        OBSERVATION|       3|\n",
      "+-------------------+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "filtered_df.groupBy(\"basisOfRecord\").count().orderBy(desc(\"count\")).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### HUMAN_OBSERVATION"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Most of the `HUMAN_OBSERVATION` are identified as citizen science"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 178,
   "metadata": {},
   "outputs": [],
   "source": [
    "human_observation_unmatched_df = (\n",
    "    spark_df\n",
    "    .filter(col(\"basisOfRecord\")==\"HUMAN_OBSERVATION\")\n",
    "    .join(\n",
    "        broadcast(matching_df), \n",
    "        matching_df._id == spark_df.datasetKey,\n",
    "        how = \"left_anti\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 180,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------+------------------------------------+-------+---------+\n",
      "|publisher                                            |datasetKey                          |count  |bucket   |\n",
      "+-----------------------------------------------------+------------------------------------+-------+---------+\n",
      "|Xeno-canto Foundation for Nature Sounds              |b1047888-ae52-4179-9dd5-5448ea342a24|1404870|1m-5m    |\n",
      "|United States Geological Survey                      |cca13f2c-0d2c-4c2f-93b9-4446c0cc1629|492239 |100k-500k|\n",
      "|Natural History Museum Rotterdam                     |6db2a74e-98c5-4be3-ae30-3ec8dc68b0f4|284982 |100k-500k|\n",
      "|Vermont Center for Ecostudies                        |cf3bdc30-370c-48d3-8fff-b587a39d72d6|264469 |100k-500k|\n",
      "|Senckenberg                                          |e5774d90-9f01-42bb-a747-32331be82b18|74534  |50k-100k |\n",
      "|India Biodiversity Portal                            |c6b86c40-ff71-4e5e-902c-111f400d0d56|66174  |50k-100k |\n",
      "|Xeno-canto Foundation for Nature Sounds              |ecce3417-d4fd-4fab-9009-09ad3f3f62b3|50590  |50k-100k |\n",
      "|GBIF Norway                                          |9807df89-7446-4aab-8ec8-fd837085c1a1|46391  |10k-50k  |\n",
      "|Estonian Ornithological Society (BirdLife Estonia)   |4b15eab0-67e6-42c6-bc1a-9bb44354d837|30921  |10k-50k  |\n",
      "|Lomonosov Moscow State University                    |bafa5083-8081-42f8-9201-04c511f45098|30742  |10k-50k  |\n",
      "|Central Michigan University Herbarium                |72c4d3c6-5b8d-49f5-bfbe-febd53849588|30083  |10k-50k  |\n",
      "|Xeno-canto Foundation for Nature Sounds              |ff571aeb-46bf-45c4-ad2c-af4d68315765|22370  |10k-50k  |\n",
      "|Atlas of Living Australia                            |b6465deb-2071-445a-9ebd-d5722e1ed7e2|21269  |10k-50k  |\n",
      "|California Department of Food and Agriculture (CDA)  |b0515413-6d32-490a-83a0-f8c08f002c70|14479  |10k-50k  |\n",
      "|Yugra State University Biological Collection (YSU BC)|2a822018-c93e-4897-852c-46ec861705af|14343  |10k-50k  |\n",
      "|Lomonosov Moscow State University                    |47cac062-7224-4570-8141-c822320627fd|13786  |10k-50k  |\n",
      "|Vanderbilt University                                |0096dfc0-9925-47ef-9700-9b77814295f1|13257  |10k-50k  |\n",
      "|Finnish Biodiversity Information Facility            |956bc674-6022-4c52-833e-c5fb39dc837a|13145  |10k-50k  |\n",
      "|Biologiezentrum Linz Oberoesterreich                 |857bce66-f762-11e1-a439-00145eb45e9a|12385  |10k-50k  |\n",
      "|Senckenberg                                          |bd63a93c-8283-11e2-b873-00145eb45e9a|11671  |10k-50k  |\n",
      "|National Museum of Natural History, Luxembourg       |9ed0adc4-0a7a-49df-9054-e26e9a7dbc8d|10237  |10k-50k  |\n",
      "|Botanic Garden and Botanical Museum Berlin           |e1beb83c-9b83-4d17-ac5e-d24e09507ec5|9942   |5k-10k   |\n",
      "|Biological Records Centre                            |3ccfe992-043a-4fff-951d-1e6e158854da|9791   |5k-10k   |\n",
      "|Meise Botanic Garden                                 |b740eaa0-0679-41dc-acb7-990d562dfa37|9603   |5k-10k   |\n",
      "|The South African Institute for Aquatic Biodiversity |1aaec653-c71c-4695-9b6e-0e26214dd817|9270   |5k-10k   |\n",
      "|Department of Biology, University of Copenhagen      |f1277dd4-84c5-4e84-b300-d5a7ac3c1edd|8552   |5k-10k   |\n",
      "|University of Oslo                                   |88d6237c-fbce-4135-b509-0aeed1e4222e|7603   |5k-10k   |\n",
      "|Staatliche Naturwissenschaftliche Sammlungen Bayerns |64dabd3c-4f34-4520-b9dd-d227a0bf1582|7109   |5k-10k   |\n",
      "|Flanders Marine Institute                            |d015c8d2-b847-4007-a401-7c4685d53218|6780   |5k-10k   |\n",
      "|National Biodiversity Centre                         |926f3dd4-89d9-4cbd-8ed7-32fb6fcfa861|6713   |5k-10k   |\n",
      "+-----------------------------------------------------+------------------------------------+-------+---------+\n",
      "only showing top 30 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "575"
      ]
     },
     "execution_count": 180,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "view_freq(\n",
    "    human_observation_unmatched_df,\n",
    "    [\"publisher\", \"datasetKey\"],\n",
    "    30,\n",
    "    truncate = False\n",
    ")\n",
    "create_freq(human_observation_unmatched_df, [\"publisher\", \"datasetKey\"]).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### MATERIAL_SAMPLE"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------------+------------------------------------+-----+-------+\n",
      "|publisher                                                  |datasetKey                          |count|bucket |\n",
      "+-----------------------------------------------------------+------------------------------------+-----+-------+\n",
      "|Natural History Museum of Denmark                          |cb8a261a-66cb-4068-809e-9e773359bb30|10846|10k-50k|\n",
      "|Miljøstyrelsen / The Danish Environmental Protection Agency|963a6b96-4d22-4428-86e4-afee52cf4a8e|1    |1-10   |\n",
      "+-----------------------------------------------------------+------------------------------------+-----+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "view_freq(\n",
    "    filtered_df.filter(col(\"basisOfRecord\").isin([\"MATERIAL_SAMPLE\"])),\n",
    "    [\"publisher\", \"datasetKey\"],\n",
    "    truncate = False\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exclude Museum Specimens & GRSciColl"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "basisOfRecord_specimen_list=[\"PRESERVED_SPECIMEN\", \"FOSSIL_SPECIMEN\", \"MATERIAL_SAMPLE\", \"LIVING_SPECIMEN\", \"MATERIAL_CITATION\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Find GRSciColl Dataset Collection Match"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+--------------------------------------------------------+------------------------------------+-----+\n",
      "|basisOfRecord     |publisher                                               |datasetKey                          |count|\n",
      "+------------------+--------------------------------------------------------+------------------------------------+-----+\n",
      "|PRESERVED_SPECIMEN|Bailey-Matthews National Shell Museum                   |417f4d21-959b-4773-90a2-c38d1822d873|3321 |\n",
      "|FOSSIL_SPECIMEN   |Virginia Museum of Natural History                      |66ca5555-102d-4d50-baaa-a949a2846666|3673 |\n",
      "|PRESERVED_SPECIMEN|Mohonk Preserve                                         |ffe1030d-42d1-4bb5-8400-1123cc859a5a|6270 |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |4cb3d289-8d1e-4831-9c01-df02084b999a|835  |\n",
      "|FOSSIL_SPECIMEN   |Virginia Museum of Natural History                      |212a4207-f330-49e7-8c84-51dccf3f1a8b|41652|\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |9a65aa81-8dc1-47b0-889d-fb34dd6826a7|552  |\n",
      "|PRESERVED_SPECIMEN|Ministerio del Medio Ambiente de Chile                  |f69687da-edff-43a7-be3a-6ec8a6f80a1d|132  |\n",
      "|PRESERVED_SPECIMEN|University of Alaska Museum of the North                |8539cd0e-f762-11e1-a439-00145eb45e9a|180  |\n",
      "|PRESERVED_SPECIMEN|Slobozhanskyi National Nature Park                      |4be164b2-8747-4b43-b7c0-2c0802496dd2|114  |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |db89acb0-b372-4f3d-9ba6-3ab8124a3214|120  |\n",
      "|PRESERVED_SPECIMEN|Instituto de Pesquisas Jardim Botanico do Rio de Janeiro|e169a84f-6726-4039-900c-2804591fcfdf|3    |\n",
      "+------------------+--------------------------------------------------------+------------------------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataset_collection_match_df = (\n",
    "    filtered_df\n",
    "    .join(\n",
    "        broadcast(grscicoll_collection_df),\n",
    "        filtered_df.collectionCode == grscicoll_collection_df.code,\n",
    "        how=\"inner\"\n",
    "    )\n",
    "    .filter(\n",
    "        col(\"basisOfRecord\").isin(basisOfRecord_specimen_list)\n",
    "    )\n",
    "    .groupBy([\"basisOfRecord\", \"publisher\", \"datasetKey\"])\n",
    "    .count()\n",
    "    .orderBy(desc(\"count\"))\n",
    "    .distinct()\n",
    ")\n",
    "dataset_collection_match_df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Find GRSciColl Dataset Institute Match"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 21:==============================================>      (672 + 96) / 768]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+--------------------------------------------------------+------------------------------------+------+\n",
      "|basisOfRecord     |publisher                                               |datasetKey                          |count |\n",
      "+------------------+--------------------------------------------------------+------------------------------------+------+\n",
      "|PRESERVED_SPECIMEN|Herbarium du CRSN-Lwiro                                 |bcffbc19-b28c-42ae-995a-ff852e158ce5|4969  |\n",
      "|PRESERVED_SPECIMEN|National Museum of Natural History, Luxembourg          |b6dc7422-a09d-4986-b5ac-e8f6f50a9732|99    |\n",
      "|PRESERVED_SPECIMEN|Butler University, Friesner Herbarium                   |900c9d31-e4a7-4e75-96a4-b701fddc1d0b|312   |\n",
      "|FOSSIL_SPECIMEN   |Virginia Museum of Natural History                      |66ca5555-102d-4d50-baaa-a949a2846666|3673  |\n",
      "|MATERIAL_SAMPLE   |Natural History Museum of Denmark                       |cb8a261a-66cb-4068-809e-9e773359bb30|174401|\n",
      "|PRESERVED_SPECIMEN|Butler University, Friesner Herbarium                   |439c0fed-d5ab-4250-9b8e-9ebdb2bb3bb5|1060  |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |4cb3d289-8d1e-4831-9c01-df02084b999a|835   |\n",
      "|FOSSIL_SPECIMEN   |Virginia Museum of Natural History                      |212a4207-f330-49e7-8c84-51dccf3f1a8b|10413 |\n",
      "|PRESERVED_SPECIMEN|Slobozhanskyi National Nature Park                      |4be164b2-8747-4b43-b7c0-2c0802496dd2|57    |\n",
      "|PRESERVED_SPECIMEN|Ministerio del Medio Ambiente de Chile                  |f69687da-edff-43a7-be3a-6ec8a6f80a1d|132   |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |9a65aa81-8dc1-47b0-889d-fb34dd6826a7|138   |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |db89acb0-b372-4f3d-9ba6-3ab8124a3214|40    |\n",
      "|PRESERVED_SPECIMEN|Instituto de Pesquisas Jardim Botanico do Rio de Janeiro|e169a84f-6726-4039-900c-2804591fcfdf|3     |\n",
      "+------------------+--------------------------------------------------------+------------------------------------+------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "dataset_institution_match_df = (\n",
    "    filtered_df\n",
    "    .join(\n",
    "        broadcast(grscicoll_institution_df),\n",
    "        filtered_df.institutionCode == grscicoll_institution_df.code,\n",
    "        how=\"inner\"\n",
    "    )\n",
    "    .filter(\n",
    "        col(\"basisOfRecord\").isin(basisOfRecord_specimen_list)\n",
    "    )\n",
    "    .groupBy([\"basisOfRecord\", \"publisher\", \"datasetKey\"])\n",
    "    .count()\n",
    "    .orderBy(desc(\"count\"))\n",
    "    .distinct()\n",
    ")\n",
    "dataset_institution_match_df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+--------------------------------------------------------+------------------------------------+------+\n",
      "|basisOfRecord     |publisher                                               |datasetKey                          |count |\n",
      "+------------------+--------------------------------------------------------+------------------------------------+------+\n",
      "|MATERIAL_SAMPLE   |Natural History Museum of Denmark                       |cb8a261a-66cb-4068-809e-9e773359bb30|174401|\n",
      "|FOSSIL_SPECIMEN   |Virginia Museum of Natural History                      |212a4207-f330-49e7-8c84-51dccf3f1a8b|41652 |\n",
      "|FOSSIL_SPECIMEN   |Virginia Museum of Natural History                      |212a4207-f330-49e7-8c84-51dccf3f1a8b|10413 |\n",
      "|PRESERVED_SPECIMEN|Mohonk Preserve                                         |ffe1030d-42d1-4bb5-8400-1123cc859a5a|6270  |\n",
      "|PRESERVED_SPECIMEN|Herbarium du CRSN-Lwiro                                 |bcffbc19-b28c-42ae-995a-ff852e158ce5|4969  |\n",
      "|FOSSIL_SPECIMEN   |Virginia Museum of Natural History                      |66ca5555-102d-4d50-baaa-a949a2846666|3673  |\n",
      "|PRESERVED_SPECIMEN|Bailey-Matthews National Shell Museum                   |417f4d21-959b-4773-90a2-c38d1822d873|3321  |\n",
      "|PRESERVED_SPECIMEN|Butler University, Friesner Herbarium                   |439c0fed-d5ab-4250-9b8e-9ebdb2bb3bb5|1060  |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |4cb3d289-8d1e-4831-9c01-df02084b999a|835   |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |9a65aa81-8dc1-47b0-889d-fb34dd6826a7|552   |\n",
      "|PRESERVED_SPECIMEN|Butler University, Friesner Herbarium                   |900c9d31-e4a7-4e75-96a4-b701fddc1d0b|312   |\n",
      "|PRESERVED_SPECIMEN|University of Alaska Museum of the North                |8539cd0e-f762-11e1-a439-00145eb45e9a|180   |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |9a65aa81-8dc1-47b0-889d-fb34dd6826a7|138   |\n",
      "|PRESERVED_SPECIMEN|Ministerio del Medio Ambiente de Chile                  |f69687da-edff-43a7-be3a-6ec8a6f80a1d|132   |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |db89acb0-b372-4f3d-9ba6-3ab8124a3214|120   |\n",
      "|PRESERVED_SPECIMEN|Slobozhanskyi National Nature Park                      |4be164b2-8747-4b43-b7c0-2c0802496dd2|114   |\n",
      "|PRESERVED_SPECIMEN|National Museum of Natural History, Luxembourg          |b6dc7422-a09d-4986-b5ac-e8f6f50a9732|99    |\n",
      "|PRESERVED_SPECIMEN|Slobozhanskyi National Nature Park                      |4be164b2-8747-4b43-b7c0-2c0802496dd2|57    |\n",
      "|PRESERVED_SPECIMEN|Virginia Museum of Natural History                      |db89acb0-b372-4f3d-9ba6-3ab8124a3214|40    |\n",
      "|PRESERVED_SPECIMEN|Instituto de Pesquisas Jardim Botanico do Rio de Janeiro|e169a84f-6726-4039-900c-2804591fcfdf|3     |\n",
      "+------------------+--------------------------------------------------------+------------------------------------+------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dataset_grscicoll_match_df = dataset_institution_match_df.union(dataset_collection_match_df).distinct().orderBy(desc(\"count\"))\n",
    "dataset_grscicoll_match_df.show(50, truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Find GRSciColl Records Match"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Collection \n",
    "record_collection_match_df = (\n",
    "    filtered_df\n",
    "    .join(\n",
    "        broadcast(grscicoll_collection_df),\n",
    "        filtered_df.collectionCode == grscicoll_collection_df.code,\n",
    "        how=\"inner\"\n",
    "    )\n",
    "    .filter(\n",
    "        col(\"basisOfRecord\").isin(basisOfRecord_specimen_list)\n",
    "    )\n",
    "    .select(\"uuid\")\n",
    "    .distinct()\n",
    ")\n",
    "\n",
    "# Institution\n",
    "record_institution_match_df = (\n",
    "    filtered_df\n",
    "    .join(\n",
    "        broadcast(grscicoll_institution_df),\n",
    "        filtered_df.institutionCode == grscicoll_institution_df.code,\n",
    "        how=\"inner\"\n",
    "    )\n",
    "    .filter(\n",
    "        col(\"basisOfRecord\").isin(basisOfRecord_specimen_list)\n",
    "    )\n",
    "    .select(\"uuid\")\n",
    "    .distinct()\n",
    ")\n",
    "\n",
    "record_specimen_match_df = record_collection_match_df.union(record_institution_match_df).distinct()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Remove Matched Datasets & Records"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "filtered_df_x_grscicoll = (\n",
    "    filtered_df\n",
    "    .join(\n",
    "        broadcast(record_specimen_match_df),\n",
    "        on = \"uuid\",\n",
    "        how=\"left_anti\"\n",
    "    )\n",
    "    .join(\n",
    "        broadcast(dataset_grscicoll_match_df.select(\"datasetKey\")),\n",
    "        on = \"datasetKey\",\n",
    "        how = \"left_anti\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 88:=====================================================>(762 + 6) / 768]68]]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+---------+---------+\n",
      "|basisOfRecord      |count    |bucket   |\n",
      "+-------------------+---------+---------+\n",
      "|HUMAN_OBSERVATION  |179198863|10m+     |\n",
      "|PRESERVED_SPECIMEN |122674   |100k-500k|\n",
      "|MACHINE_OBSERVATION|48093    |10k-50k  |\n",
      "|OBSERVATION        |17002    |10k-50k  |\n",
      "|OCCURRENCE         |1859     |1k-5k    |\n",
      "|MATERIAL_SAMPLE    |4        |1-10     |\n",
      "|FOSSIL_SPECIMEN    |1        |1-10     |\n",
      "+-------------------+---------+---------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "view_freq(filtered_df_x_grscicoll, \"basisOfRecord\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                768]]]\r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "179388496"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filtered_df_x_grscicoll.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Generate Lookup Table for Multi-Images Subset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Check if there is an active Spark session\n",
    "spark= SparkSession.getActiveSession()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                / 768]\r"
     ]
    }
   ],
   "source": [
    "(\n",
    "    filtered_df_x_grscicoll\n",
    "    .repartition(10)\n",
    "    .write\n",
    "    .mode(\"overwrite\")\n",
    "    .parquet(f\"{BASE_PATH}/gbif/attributes/occurrence_citizen_science\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "filtered_df_x_grscicoll = spark.read.parquet(f\"{BASE_PATH}/gbif/attributes/occurrence_citizen_science\")\n",
    "lookup_tbl = spark.read.parquet(f\"{BASE_PATH}/gbif/lookup_tables/2024-05-01/lookup_tables\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "lookup_tbl_citizen_science = (\n",
    "    filtered_df_x_grscicoll\n",
    "    .join(source_id_with_multiple_uuids.select(\"source_id\"), on=\"source_id\", how=\"inner\")\n",
    "    .select([\"uuid\", \"source_id\", \"basisOfRecord\", \"publisher\", \"datasetKey\", \"scientificName\", \"taxonRank\"] + COLS_TAXONOMIC)\n",
    "    .join(lookup_tbl, on=\"uuid\", how=\"inner\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark_df = spark.read.parquet(f\"{BASE_PATH}/gbif/attributes/cols_of_interest\")\n",
    "\n",
    "filtered_df = (\n",
    "    spark_df\n",
    "    .join(\n",
    "        filtered_df_x_grscicoll.select(\"uuid\"),\n",
    "        on = \"uuid\",\n",
    "        how = \"inner\"\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+----------------------+--------------------+---------------------+---------------------+----------------------+----------+--------------------+\n",
      "|sum_taxon_higher_rank|mean_taxon_higher_rank|sum_taxon_match_none|mean_taxon_match_none|sum_taxon_match_fuzzy|mean_taxon_match_fuzzy|sum_is_any|mean_is_any         |\n",
      "+---------------------+----------------------+--------------------+---------------------+---------------------+----------------------+----------+--------------------+\n",
      "|2383044              |0.013284263222765411  |81029               |4.516956315860968E-4 |690920               |0.003851529030044379  |3154993   |0.017587487884395885|\n",
      "+---------------------+----------------------+--------------------+---------------------+---------------------+----------------------+----------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import col, when, mean, sum as spark_sum\n",
    "\n",
    "# Generate boolean columns and cast to integers for aggregation\n",
    "spark_df_with_flags = filtered_df.withColumn(\n",
    "    \"is_taxon_higher_rank\", \n",
    "    when(col(\"issue\").contains(\"TAXON_MATCH_HIGHERRANK\"), 1).otherwise(0)\n",
    ").withColumn(\n",
    "    \"is_taxon_match_none\", \n",
    "    when(col(\"issue\").contains(\"TAXON_MATCH_NONE\"), 1).otherwise(0)\n",
    ").withColumn(\n",
    "    \"is_taxon_match_fuzzy\", \n",
    "    when(col(\"issue\").contains(\"TAXON_MATCH_FUZZY\"), 1).otherwise(0)\n",
    ").withColumn(\n",
    "    \"is_any\", \n",
    "    when((col(\"issue\").contains(\"TAXON_MATCH_FUZZY\")) | (col(\"issue\").contains(\"TAXON_MATCH_NONE\")) | (col(\"issue\").contains(\"TAXON_MATCH_HIGHERRANK\")), 1).otherwise(0)\n",
    ")\n",
    "\n",
    "# Calculate sum and mean for each numeric column\n",
    "result_df = spark_df_with_flags.agg(\n",
    "    spark_sum(\"is_taxon_higher_rank\").alias(\"sum_taxon_higher_rank\"),\n",
    "    mean(\"is_taxon_higher_rank\").alias(\"mean_taxon_higher_rank\"),\n",
    "    spark_sum(\"is_taxon_match_none\").alias(\"sum_taxon_match_none\"),\n",
    "    mean(\"is_taxon_match_none\").alias(\"mean_taxon_match_none\"),\n",
    "    spark_sum(\"is_taxon_match_fuzzy\").alias(\"sum_taxon_match_fuzzy\"),\n",
    "    mean(\"is_taxon_match_fuzzy\").alias(\"mean_taxon_match_fuzzy\"),\n",
    "    spark_sum(\"is_any\").alias(\"sum_is_any\"),\n",
    "    mean(\"is_any\").alias(\"mean_is_any\")\n",
    ")\n",
    "\n",
    "result_df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## N_MAX_FILES: 100"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "N_MAX_FILES = 100\n",
    "\n",
    "unique_paths = lookup_tbl_citizen_science.select(\"path\").distinct()\n",
    "window_spec = Window.orderBy(\"path\")\n",
    "unique_paths_with_row = unique_paths.withColumn(\n",
    "    \"row_number\", row_number().over(window_spec)\n",
    ")\n",
    "\n",
    "grouped_paths = unique_paths_with_row.withColumn(\n",
    "    \"group_id\", floor((col(\"row_number\") - 1) / N_MAX_FILES)\n",
    ").drop(\"row_number\")\n",
    "\n",
    "result_lookup_tbl = (\n",
    "    lookup_tbl_citizen_science\n",
    "    .join(grouped_paths, on=\"path\", how=\"left\")\n",
    "    .select([\"uuid\", \"source_id\", \"basisOfRecord\", \"publisher\", \"datasetKey\", \"scientificName\", \"taxonRank\"] + COLS_TAXONOMIC+ [\"path\", \"group_id\"])\n",
    "    .repartition(1, \"group_id\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "(\n",
    "    result_lookup_tbl\n",
    "    .write.partitionBy(\"group_id\").mode(\"overwrite\")\n",
    "    .parquet(f\"{BASE_PATH}/gbif/lookup_tables/2024-05-01/lookup_multi_images_citizen_science\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- uuid: string (nullable = true)\n",
      " |-- source_id: string (nullable = true)\n",
      " |-- basisOfRecord: string (nullable = true)\n",
      " |-- publisher: string (nullable = true)\n",
      " |-- datasetKey: string (nullable = true)\n",
      " |-- scientificName: string (nullable = true)\n",
      " |-- taxonRank: string (nullable = true)\n",
      " |-- kingdom: string (nullable = true)\n",
      " |-- phylum: string (nullable = true)\n",
      " |-- class: string (nullable = true)\n",
      " |-- order: string (nullable = true)\n",
      " |-- family: string (nullable = true)\n",
      " |-- genus: string (nullable = true)\n",
      " |-- species: string (nullable = true)\n",
      " |-- path: string (nullable = true)\n",
      " |-- group_id: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "result_lookup_tbl.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## N_MAX_FILES: 300"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "N_MAX_FILES = 300\n",
    "\n",
    "unique_paths = lookup_tbl_citizen_science.select(\"path\").distinct()\n",
    "window_spec = Window.orderBy(\"path\")\n",
    "unique_paths_with_row = unique_paths.withColumn(\n",
    "    \"row_number\", row_number().over(window_spec)\n",
    ")\n",
    "\n",
    "grouped_paths = unique_paths_with_row.withColumn(\n",
    "    \"group_id\", floor((col(\"row_number\") - 1) / N_MAX_FILES)\n",
    ").drop(\"row_number\")\n",
    "\n",
    "result_lookup_tbl = (\n",
    "    lookup_tbl_citizen_science\n",
    "    .join(grouped_paths, on=\"path\", how=\"left\")\n",
    "    .select([\"uuid\", \"source_id\", \"basisOfRecord\", \"publisher\", \"datasetKey\", \"scientificName\", \"taxonRank\"] + COLS_TAXONOMIC+ [\"path\", \"group_id\"])\n",
    "    .repartition(1, \"group_id\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "(\n",
    "    result_lookup_tbl\n",
    "    .write.partitionBy(\"group_id\").mode(\"overwrite\")\n",
    "    .parquet(f\"{BASE_PATH}/gbif/lookup_tables/2024-05-01/lookup_multi_images_citizen_science_300\")\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def process_group(spark, base_input_path, base_output_path, group_id):\n",
    "\n",
    "    # Construct paths for the current group\n",
    "    group_input_path = f\"{base_input_path}/group_id={group_id}\"\n",
    "    group_output_path = f\"{base_output_path}/group_id={group_id}\"\n",
    "\n",
    "\n",
    "    filtered_df = spark.read.parquet(group_input_path)\n",
    "    unique_paths = [row['path'] for row in filtered_df.select(\"path\").distinct().collect()]\n",
    "    \n",
    "    # Read the combined DataFrame from unique paths\n",
    "    combined_df = spark.read.parquet(*unique_paths).select([\"uuid\", \"original_size\", \"resized_size\", \"image\"])\n",
    "\n",
    "    result_df = combined_df.join(broadcast(filtered_df), on=\"uuid\", how=\"inner\")\n",
    "    result_df = result_df.dropDuplicates([\"uuid\"]).repartition(100)\n",
    "    \n",
    "    # Write the result to the output path\n",
    "    result_df.write.mode(\"overwrite\").parquet(group_output_path)\n",
    "    print(f\"Processed and saved results for group_id={group_id} to {group_output_path}\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "base_input_path = f\"{BASE_PATH}/gbif/lookup_tables/2024-05-01/lookup_multi_images_citizen_science\"\n",
    "base_output_path = f\"{BASE_PATH}/gbif/image_lookup/multi_images_citizen_science\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "def init_spark() -> SparkSession:\n",
    "    spark = (SparkSession.builder\n",
    "             .appName(\"GBIF EDA\")\n",
    "             .config(\"spark.executor.instances\", \"80\")\n",
    "             .config(\"spark.executor.memory\", \"75G\")\n",
    "             .config(\"spark.executor.cores\", \"12\")\n",
    "             .config(\"spark.sql.parquet.enableVectorizedReader\", \"false\") \n",
    "             .getOrCreate())\n",
    "    \n",
    "    return spark\n",
    "\n",
    "spark = init_spark()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "process_group(spark, base_input_path, base_output_path, \"0\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Comparison & Summary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------------+---------+------------+------------------------+\n",
      "|basisOfRecord      |n_occurrence|n_CS_nlp |n_CS_matched|n_CS_matched_x_grscicoll|\n",
      "+-------------------+------------+---------+------------+------------------------+\n",
      "|HUMAN_OBSERVATION  |181826921   |178283233|179199017   |179199017               |\n",
      "|PRESERVED_SPECIMEN |58611705    |384164   |136773      |122678                  |\n",
      "|MATERIAL_CITATION  |1492        |null     |null        |null                    |\n",
      "|OBSERVATION        |35199       |17002    |17002       |17002                   |\n",
      "|MACHINE_OBSERVATION|3044865     |93040    |48093       |48093                   |\n",
      "|MATERIAL_SAMPLE    |3272814     |4        |174405      |4                       |\n",
      "|OCCURRENCE         |92770       |1995     |1859        |1859                    |\n",
      "|LIVING_SPECIMEN    |99740       |36       |null        |null                    |\n",
      "|FOSSIL_SPECIMEN    |2001234     |1        |14087       |1                       |\n",
      "+-------------------+------------+---------+------------+------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "filtered_df_nlp = (\n",
    "    spark_df\n",
    "    .join(\n",
    "        broadcast(matching_df_nlp),\n",
    "        on = \"datasetKey\",\n",
    "        how=\"inner\"\n",
    "    )\n",
    ")\n",
    "(\n",
    "    create_freq(spark_df, \"basisOfRecord\").selectExpr(\"basisOfRecord\", \"count AS n_occurrence\")\n",
    "    .join(\n",
    "        create_freq(filtered_df_nlp, \"basisOfRecord\").selectExpr(\"basisOfRecord\", \"count AS n_CS_nlp\"),\n",
    "        on = \"basisOfRecord\",\n",
    "        how=\"left\"\n",
    "    )\n",
    "    .join(\n",
    "        create_freq(filtered_df, \"basisOfRecord\").selectExpr(\"basisOfRecord\", \"count AS n_CS_matched\"),\n",
    "        on = \"basisOfRecord\",\n",
    "        how=\"left\"\n",
    "    )\n",
    "    .join(\n",
    "        create_freq(filtered_df_x_grscicoll, \"basisOfRecord\").selectExpr(\"basisOfRecord\", \"count AS n_CS_matched_x_grscicoll\"),\n",
    "        on = \"basisOfRecord\",\n",
    "        how=\"left\"\n",
    "    )\n",
    ").show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Occurrence dataset count: 1873\n",
      "NLP dataset count: 63\n",
      "Full-text search dataset count: 107\n",
      "Full-text search dataset count (exclude grscicoll): 93\n"
     ]
    }
   ],
   "source": [
    "print(f\"Occurrence dataset count: {spark_df.select('datasetKey').distinct().count()}\")\n",
    "\n",
    "print(f\"NLP dataset count: {filtered_df_nlp.select('datasetKey').distinct().count()}\")\n",
    "\n",
    "print(f\"Full-text search dataset count: {filtered_df.select('datasetKey').distinct().count()}\")\n",
    "\n",
    "print(f\"Full-text search dataset count (exclude grscicoll): {filtered_df_x_grscicoll.select('datasetKey').distinct().count()}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Occurrence `source_id` count: 169112381\n",
      "NLP `source_id` count: 109356415\n",
      "Full-text search `source_id` count: 110323916\n",
      "Full-text search `source_id` count (exclude grscicoll): 110286547\n"
     ]
    }
   ],
   "source": [
    "print(f\"Occurrence `source_id` count: {spark_df.select('source_id').distinct().count()}\")\n",
    "\n",
    "print(f\"NLP `source_id` count: {filtered_df_nlp.select('source_id').distinct().count()}\")\n",
    "\n",
    "print(f\"Full-text search `source_id` count: {filtered_df.select('source_id').distinct().count()}\")\n",
    "\n",
    "print(f\"Full-text search `source_id` count (exclude grscicoll): {filtered_df_x_grscicoll.select('source_id').distinct().count()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| **Type**                                | **Dataset Count** | **`source_id` Count** |\n",
    "|-----------------------------------------|------------------:|----------------------:|\n",
    "| Occurrence                              |              1873 |            169,112,381 |\n",
    "| NLP                                     |                63 |            109,356,415 |\n",
    "| Full-text search                        |               107 |            110,323,916 |\n",
    "| Full-text search (exclude grscicoll)    |                93 |            110,286,547 |\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 183,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------------------------------------------+-------------------+-----+--------+\n",
      "|publisher                                                  |basisOfRecord      |count|bucket  |\n",
      "+-----------------------------------------------------------+-------------------+-----+--------+\n",
      "|SLU Artdatabanken                                          |PRESERVED_SPECIMEN |50220|50k-100k|\n",
      "|Finnish Biodiversity Information Facility                  |MACHINE_OBSERVATION|47024|10k-50k |\n",
      "|The Norwegian Biodiversity Information Centre (NBIC)       |PRESERVED_SPECIMEN |39004|10k-50k |\n",
      "|ClimateWatch                                               |OBSERVATION        |17002|10k-50k |\n",
      "|University of Texas at Austin, Biodiversity Collections    |PRESERVED_SPECIMEN |11855|10k-50k |\n",
      "|Danish Mycological Society                                 |PRESERVED_SPECIMEN |11300|10k-50k |\n",
      "|Finnish Biodiversity Information Facility                  |PRESERVED_SPECIMEN |9257 |5k-10k  |\n",
      "|myFOSSIL eMuseum                                           |FOSSIL_SPECIMEN    |3180 |1k-5k   |\n",
      "|Atlas of Living Australia                                  |OCCURRENCE         |1859 |1k-5k   |\n",
      "|Miljøstyrelsen / The Danish Environmental Protection Agency|PRESERVED_SPECIMEN |1030 |1k-5k   |\n",
      "|Miljøstyrelsen / The Danish Environmental Protection Agency|MACHINE_OBSERVATION|969  |501-1k  |\n",
      "|Jardín Botánico de Bogotá \"José Celestino Mutis\"           |MACHINE_OBSERVATION|62   |51-100  |\n",
      "|Negrita Films SAS                                          |MACHINE_OBSERVATION|38   |11-50   |\n",
      "|Regional historical, cultural and ecological center        |PRESERVED_SPECIMEN |7    |1-10    |\n",
      "|Miljøstyrelsen / The Danish Environmental Protection Agency|MATERIAL_SAMPLE    |4    |1-10    |\n",
      "|Herbarium du CRSN-Lwiro                                    |PRESERVED_SPECIMEN |4    |1-10    |\n",
      "|National Museum of Natural History, Luxembourg             |PRESERVED_SPECIMEN |1    |1-10    |\n",
      "|Miljøstyrelsen / The Danish Environmental Protection Agency|FOSSIL_SPECIMEN    |1    |1-10    |\n",
      "+-----------------------------------------------------------+-------------------+-----+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "view_freq(\n",
    "    filtered_df_x_grscicoll.filter(\n",
    "        (col(\"basisOfRecord\") != \"HUMAN_OBSERVATION\")\n",
    "    ),\n",
    "    [\"publisher\", \"basisOfRecord\"],\n",
    "    truncate = False\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "110286389"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filtered_df_x_grscicoll.select(\"source_id\").distinct().count()"
   ]
  }
 ],
 "metadata": {
  "jupytext": {
   "formats": "ipynb,py:percent"
  },
  "kernelspec": {
   "display_name": "PySpark",
   "language": "python",
   "name": "pyspark"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
